/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2;

import com.skjegstad.mist2.datatypes.*;
import com.skjegstad.mist2.events.NewDataElementReceivedEvent;
import com.skjegstad.mist2.events.NewSubscriptionEvent;
import com.skjegstad.mist2.events.RemoveDataElementsEvent;
import com.skjegstad.mist2.events.listeners.LogEventListener;
import com.skjegstad.mist2.events.listeners.OutputStreamExceptionEventListener;
import com.skjegstad.mist2.exceptions.MistException;
import com.skjegstad.mist2.exceptions.MistHashCollisionException;
import com.skjegstad.mist2.managers.DataStoreManager;
import com.skjegstad.mist2.managers.KnownNodesStoreManager;
import com.skjegstad.mist2.managers.MistEventManager;
import com.skjegstad.mist2.modules.tcp.TCPNetworkModule;
import com.skjegstad.mist2.modules.udp.broadcast.UDPBroadcastModule;
import com.skjegstad.mist2.modules.udp.unicast.UDPUnicastModule;
import com.skjegstad.mist2.storage.KnownNodesStore;
import com.skjegstad.simpleevents.SimpleEvent;
import com.skjegstad.simpleevents.SimpleEventListener;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Map.Entry;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 *
 * @author Magnus Skjegstad
 */
public class MistCoreManager {

    private final ConcurrentMap<ConnectionIdentifier, Integer> connectionFragmentSize =
            new ConcurrentHashMap<ConnectionIdentifier, Integer>();

    private final ConcurrentMap<ConnectionIdentifier, List<TopicIdentifier>> topicsAllowedPerConnection =
            new ConcurrentHashMap();

    private DataStoreManager dataStoreManager;
    private KnownNodesStoreManager knownNodesStoreManager;
    private final MistEventManager eventManager = new MistEventManager();
    private List<IMistNetworkModule> networkModules = new ArrayList();

    private ConcurrentMap<TopicIdentifier, List<IMistSubscriptionListener>> topicListeners;

    private List<TopicIdentifier> myTopics;
    private int defaultTopicTtl = 5;
    private int maxTopicTtlAccepted = defaultTopicTtl;
    private byte maximumExtraBloomfilterBits = 64;
    private int maximumDataElementAge = 3600 * 1000;
    private NodeIdentifier myNodeId;
    private int seqNr;
    private final Random r = new Random();

    private UDPUnicastModule udpUnicastModule;
    private TCPNetworkModule tcpModule;
    
    private boolean isStarted = false;

    public MistCoreManager(long nodeId) {
        this.myTopics = new ArrayList();
        if (nodeId >= 0)
            this.myNodeId = new NodeIdentifier(nodeId);
        else
            this.myNodeId = new NodeIdentifier(Math.abs(r.nextLong()));
        
        seqNr = r.nextInt();
    }

    public MistCoreManager() {
        this(-1);
    }

    public boolean isDebug() {
        return MistUtil.getDebug();
    }

    public void setDebug(boolean debug) {
        MistUtil.setDebug(debug);
    }
    

    /**
     * Add network module. Module must implement IMistNetworkModule and must
     * be added before start() is called.
     *
     * @param networkModule network module.
     * @param routablePrefixes topics that matches this prefix will be routed from other network modules to this one. If null, all topics are routed.
     */
    private void addNetworkModule(IMistNetworkModule networkModule, List<TopicIdentifier> routablePrefixes) throws MistException {
        networkModules.add(networkModule);
        networkModule.setConnectionIdentifier(new ConnectionIdentifier());
        if (routablePrefixes != null)
            topicsAllowedPerConnection.put(networkModule.getConnectionIdentifier(), routablePrefixes);
        if (isStarted)
            startNetworkModules(true);
    }

    private boolean prefixMatchTopic(List<TopicIdentifier> prefixes, TopicIdentifier topicToMatch) {
        if (prefixes == null)
            return true;

        for (TopicIdentifier prefix : prefixes)
            if (topicToMatch.prefixMatch(prefix))
                return true;

        return false;
    }

    public TopicMap getActiveTopics(ConnectionIdentifier connectionIdentifier) {
        // get routable topics from all nodes + myTopics
        TopicMap activeTopics = new TopicMap();

        // get prefixes
        List<TopicIdentifier> prefixes = topicsAllowedPerConnection.get(connectionIdentifier);

        // store own topics
        for (TopicIdentifier t : myTopics)
            if (prefixMatchTopic(prefixes, t))
                activeTopics.put(t, defaultTopicTtl, getMyNodeId().getNodeIdentifier());

        // get all nodes
        for (KnownNodesStore n : knownNodesStoreManager.getKnownNodesForAllConnections()) {
            TopicMap knownTopics = n.getAllKnownTopicsWithTtl();
            if (knownTopics != null && !knownTopics.isEmpty())
                // loop through topics for each node
                for (Entry<TopicIdentifier, TopicMapEntry> topicWithTtl : knownTopics.entrySet()) {
                    // add topic with decreased ttl
                    TopicIdentifier newTopic = topicWithTtl.getKey();
                    int newTtl = topicWithTtl.getValue().getTtl()-1;
                    
                    if (newTtl <= 0) // skip topics with ttl 0
                        continue; 
                    
                    if (newTtl >= maxTopicTtlAccepted) // skip topics with max ttl
                        continue;

                    if ((activeTopics.get(newTopic) != null) && (newTtl <= activeTopics.get(newTopic).getTtl())) // skip if topic exists with higher ttl
                        continue;

                    if (!prefixMatchTopic(prefixes, newTopic)) // skip topics without matching prefix
                        continue;

                    // ok, add topic and xor with our own node id
                    activeTopics.put(newTopic, newTtl, 
                            topicWithTtl.getValue().getChecksum() ^ getMyNodeId().getNodeIdentifier());
                }
        }

        return activeTopics;
    }

    /**
     * Start all network modules added with addNetworkModule() by invoking their start()-methods. Called from start().
     * 
     * @throws MistException if one of the modules fails to start.
     */
    protected void startNetworkModules(boolean throwException) throws MistException {
        for (IMistNetworkModule module : networkModules) {
            ConnectionIdentifier id = module.getConnectionIdentifier();
            if (connectionFragmentSize.get(id) == null) {
                connectionFragmentSize.put(id, module.getFragmentSize());
                try {
                    module.start(new NetworkModuleHelper(id, module.getFragmentSize(), this, eventManager));
                } catch (MistException ex) {
                    if (!throwException)
                        eventManager.throwException(new MistException("Unable to start " + module.getClass().getName(), ex));
                    else
                        throw new MistException("Network module failed", ex);
                }
            }
        }
    }

    /**
     * Stop all network modules added with addNetworkModule() by invoking their stop()-methods. Called from stop().
     */
    protected void stopNetworkModules() {       
        for (IMistNetworkModule module : networkModules) {
            eventManager.log("Stopping network module " + module.getClass().getSimpleName());
            try {
                module.stop();
            } catch (Exception ex) {
                eventManager.log("Got exception while stopping " + module.getClass().getSimpleName() + ". Module may be in an unstable state.");
                eventManager.throwException(new MistException("Unable to stop " + module.getClass().getSimpleName(), ex));
            }
        }
    }

    /**
     * Start Mist2 and sub-modules.
     * 
     * @throws MistException on error.
     */
    public void start() throws MistException {
        // Register exception handler
        eventManager.getExceptionPipe().listen(new OutputStreamExceptionEventListener(getMyNodeId()));
        eventManager.getLogPipe().listen(new LogEventListener(getMyNodeId()));

        dataStoreManager = new DataStoreManager(eventManager);
        knownNodesStoreManager = new KnownNodesStoreManager(getMyNodeId(), eventManager);

        // Register event listeners
        eventManager.getDataManagerEventPipe().listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof NewDataElementReceivedEvent) {
                    NewDataElementReceivedEvent e = (NewDataElementReceivedEvent) event;
                    TopicIdentifier receivedTopic = e.getDataElement().getTopic();
                    HashSet<IMistSubscriptionListener> list = null;

                    if (topicListeners != null && !topicListeners.isEmpty() && receivedTopic != null) {
                            list = new HashSet();
                            // add wildcard matches
                            for (Entry<TopicIdentifier, List<IMistSubscriptionListener>> entry : topicListeners.entrySet()) { // go through topics
                                if (receivedTopic.matchWildcard(entry.getKey())) // match on wildcard 
                                    list.addAll(entry.getValue()); // add all listeners
                            }
                    }
                    if (list == null || list.isEmpty())
                        MistUtil.debug("Received message matched none of our subscriptions");
                    else 
                        MistUtil.debug("Received message matched " + list.size() + " of our subscriptions");

                    if (list != null && !list.isEmpty())
                            for (IMistSubscriptionListener m : list)
                                try {
                                    m.deliverMessage(e.getDataElement());
                                } catch (MistException ex) {
                                    ex.printStackTrace(System.err);
                                }        
                }
            }
        });
        
        // Start network modules
        startNetworkModules(true);
        
        isStarted = true;
    }

    /**
     * Stop mist2 and sub modules.
     */
    public void stop() {
        stopNetworkModules();
        eventManager.shutdown();
        isStarted = false;
    }

    /**
     * Get active instance of DataStoreManager. Should be used directly by sub-modules. Use one of
     * the helper-classes or trigger events instead.
     * 
     * @return DataStoreManager
     */
    public DataStoreManager getDataStoreManager() {
        return dataStoreManager;
    }

    /**
     * Get active instance of KnownNodesStoreManager. Should not be used directly by sub-modules. Use one of
     * the helper-classes or trigger events instead.
     *
     * @return KnownNodesStoreManager
     */
    public KnownNodesStoreManager getKnownNodesStoreManager() {
        return knownNodesStoreManager;
    }

    /**
     * Get current instance of event manager.
     *
     * @return
     */
    public MistEventManager getEventManager() {
        return this.eventManager;
    }

    /**
     * Get maximum TTL used for topics.
     * 
     * @return
     */
    public int getMaxTopicTtl() {
        return defaultTopicTtl;
    }

    /**
     * Get topics we have subscribed to.
     * 
     * @return
     */
    public List<TopicIdentifier> getMyTopics() {
        return Collections.unmodifiableList(myTopics);
    }

    /**
     * Get list of network modules.
     * @return
     */
    public List<IMistNetworkModule> getNetworkModules() {
        return Collections.unmodifiableList(networkModules);
    }

    /**
     * Get current node Id. This is randomly generated when the class is instantiated.
     * 
     * @return current node identifier.
     */
    public NodeIdentifier getMyNodeId() {
        return myNodeId;
    }

    public Collection<DataElement> getDataElements() {
        return dataStoreManager.getDataElements(dataStoreManager.getDataElementHashes());
    }

    /**
     * Add new data element to data store and notify all modules. Triggers
     * a RegisterDataElementReceivedEvent. If the data element is new, a
     * NewDataElementReceivedEvent is triggered as well.
     *
     * @param element
     */
    public void putDataElement(DataElement element) throws MistHashCollisionException {
        dataStoreManager.registerDataElement(null, element);
    }

    public DataElement publish(NodeIdentifier creatorId, long seqNr, TopicIdentifier topic, byte[] data, long expireTimeInMs) throws MistHashCollisionException {
        DataElement element = new DataElement(creatorId, seqNr, data, topic, 0, expireTimeInMs);
        this.putDataElement(element);
        return element;
    }

    public DataElement publish(TopicIdentifier topic, byte[] data, long expireTimeInMs) throws MistHashCollisionException {
        return publish(getMyNodeId(), seqNr++, topic, data, expireTimeInMs);
    }

    public DataElement publish(String topic, byte[] data, long expireTimeInMs) throws MistHashCollisionException {
        return this.publish(new TopicIdentifier(topic), data, expireTimeInMs);
    }

    public DataElement publish(TopicIdentifier topic, byte[] data) throws MistHashCollisionException {
        return publish(topic, data, 60000);
    }

    public DataElement publish(String topic, byte[] data) throws MistHashCollisionException {
        return this.publish(new TopicIdentifier(topic), data);
    }
    
    public boolean isSubscribed(String topic) {
        TopicIdentifier t = new TopicIdentifier(topic);
        return myTopics.contains(t);            
    }

    public void subscribe(TopicIdentifier topic, IMistSubscriptionListener subscriptionListener) {
        if (!myTopics.contains(topic))
            myTopics.add(topic);

        if (topicListeners == null)
            topicListeners = new ConcurrentHashMap();
        
        if (subscriptionListener != null) {
            List<IMistSubscriptionListener> list;
            list = topicListeners.get(topic);
            if (list == null) {
                list = Collections.synchronizedList(new ArrayList());
                topicListeners.put(topic, list);
            }
            list.add(subscriptionListener);
        }

        eventManager.getDefaultPipe().trigger(new NewSubscriptionEvent(topic));
    }    

    public void subscribe(String topic, IMistSubscriptionListener subscriptionListener) {
        TopicIdentifier t = new TopicIdentifier(topic);
        this.subscribe(t, subscriptionListener);
    }

    public void unsubscribe(String topic) {
        TopicIdentifier t = new TopicIdentifier(topic);
        this.unsubscribe(t);
    }
    
    public void unsubscribe(TopicIdentifier topic) {
        myTopics.remove(topic);
        topicListeners.remove(topic);
    }

    public void unpublish(Collection<DataElement> elementsToRemove) {
        eventManager.getDataManagerEventPipe().trigger(new RemoveDataElementsEvent(elementsToRemove));
    }

    public int getMaximumDataElementAge() {
        return this.maximumDataElementAge;
    }
    
    public byte getMaximumExtraBloomfilterBits() {
        return this.maximumExtraBloomfilterBits;
    }

    /**
     * Enable the UDP broadcast module. The module supports both UDP broadcast and multicast. See also
     * the helper methods enableMulticast() and enableBroadcast().
     *
     * @param address the address to use for broadcast or multicast.
     * @param port listening port
     * @param useMulticast enable multicast. If false, UDP broadcast is used.
     * @param config optional configuration properties.
     * @param routablePrefixes topic prefixes that should be routed to this module. Null is all topics.
     *
     */
    public void addUDPBroadcastModule(InetAddress address, int port, boolean useMulticast, Properties config, List<TopicIdentifier> routablePrefixes) throws MistException {
        this.addNetworkModule(new UDPBroadcastModule(address, port, useMulticast, config), routablePrefixes);
    }

    /**
     * Enable the multicast module. Fragment size is set to approx. 64k.
     *
     * @param multicastAddress multicast group address to use.
     * @param port port to listen/send on.
     * @param networkInterface network interface to use for multicast.
     */
    public void enableMulticast(InetAddress multicastAddress, int port, NetworkInterface networkInterface, int periodicBeaconInterval) throws MistException {
        Properties config = new Properties();
        if (networkInterface != null)
            config.setProperty("networkInterfaceName", networkInterface.getName());
        if (periodicBeaconInterval > 0)
            config.setProperty("periodicBeaconTime", String.valueOf(periodicBeaconInterval));
        this.addUDPBroadcastModule(multicastAddress, port, true, config, null);
    }
    
    public void enableMulticast(InetAddress multicastAddress, int port, NetworkInterface networkInterface) throws MistException {
        enableMulticast(multicastAddress, port, networkInterface, 0);
    }

    /**
     * Enable the multicast module. Fragment size is set to approx. 64k.
     *
     * @param multicastAddress multicast group address to use.
     * @param port port to listen/send on.
     */
    public void enableMulticast(InetAddress multicastAddress, int port) throws MistException {
        this.addUDPBroadcastModule(multicastAddress, port, true, null, null);
    }

    /**
     * Enable the multicast module. Fragment size is set to approx. 64k.
     *
     * @param multicastAddress multicast group address to use.
     * @param port port to listen/send on.
     * @param ipAddress string representation of IP address on network interface used for multicast.
     */
    public void enableMulticast(InetAddress multicastAddress, int port, String ipAddress) throws MistException {
        Properties config = new Properties();
        if (ipAddress != null)
            config.setProperty("networkInterfaceIp", ipAddress);
        this.addUDPBroadcastModule(multicastAddress, port, true, config, null);
    }


    /**
     * Enable UDP broadcast modue. Fragment size is set to approx 1400 bytes.
     *
     * @param broadcastAddress broadcast address to use, e.g. 255.255.255.255
     * @param port port to use.
     */
    public void enableBroadcast(InetAddress broadcastAddress, int port) throws MistException {
        this.addUDPBroadcastModule(broadcastAddress, port, false, null, null);
    }

    public void enableUDPUnicastServer(int port) throws MistException {
        if (udpUnicastModule != null) {
            throw new MistException("UDP unicast already enabled on port " + udpUnicastModule.getListenPort());
        }
        udpUnicastModule = new UDPUnicastModule(port);
        this.addNetworkModule(udpUnicastModule, null);
    }

    public void addUDPUnicastNeighbor(InetAddress dstAddress, int port) throws MistException {
        if (udpUnicastModule == null)
            throw new MistException("UDPUnicastServer not enabled");

        udpUnicastModule.tryToConnectToNeighbor(dstAddress, port);
    }

    public void enableTCPServer(int port) throws MistException {
        if (tcpModule != null)
            throw new MistException("TCP server already started");

        tcpModule = new TCPNetworkModule(port);
        this.addNetworkModule(tcpModule, null);
    }

    public void connectToTCPServer(InetAddress dst, int port) throws MistException {
        if (tcpModule == null)
            throw new MistException("TCP network module not enabled");
        tcpModule.connect(dst, port);
    }

    public boolean isStarted() {
        return isStarted;
    }

}
